package cn.callmee.springboot.pulsar.starter.client;

import cn.callmee.springboot.pulsar.starter.client.config.PulsarClientInitialConfiguration;
import lombok.SneakyThrows;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.TypedMessageBuilder;

import javax.annotation.Resource;
import java.util.concurrent.CompletableFuture;

@SuppressWarnings("unchecked")
public class PulsarProducerTemplate<T> {

    @Resource
    private PulsarClientInitialConfiguration pulsarClientInitialConfiguration;

    @SneakyThrows
    public MessageId send(String topic, T msg) {
        return pulsarClientInitialConfiguration.getProducer(topic)
                .getRight()
                .send(msg);
    }

    public CompletableFuture<MessageId> sendAsync(String topic, T message) {
        return pulsarClientInitialConfiguration.getProducer(topic)
                .getRight()
                .sendAsync(message);
    }

    public TypedMessageBuilder<T> createMessage(String topic, T message) {
        return pulsarClientInitialConfiguration.getProducer(topic)
                .getRight()
                .newMessage()
                .value(message);
    }
}
